#include "boost/asio/steady_timer.hpp"
#include "boost/cobalt/detached.hpp"
#include "boost/cobalt/gather.hpp"
#include "boost/cobalt/promise.hpp"
#include "boost/cobalt/task.hpp"
#include <atomic>
#include <boost/callable_traits/args.hpp>
#include <boost/cobalt.hpp>
#include <boost/signals2.hpp>
#include <chrono>
#include <cstddef>
#include <curl/curl.h>
#include <functional>
#include <iostream>
#include <map>
#include <optional>
#include <queue>
#include <string>
#include <thread>
#include <vector>

namespace cobalt = boost::cobalt;

struct request {
    CURL *handle{nullptr};
    cobalt::channel<CURLcode> *chan{nullptr};
};

namespace signals = boost::signals2;
template <typename Signal> struct signal_awaitable {
    using args_type =
        boost::callable_traits::args_t<typename Signal::signature_type>;

    bool await_ready() {
        return false;
    } // < always wait for the signal to fire.
    void await_suspend(std::coroutine_handle<void> h) {
        awaited_from.reset(h.address());
        // the handler will get copied, so we can't capture the handle with a
        // unique_ptr
        signal.connect_extended(
            [this, _ = boost::intrusive_ptr<signal_awaitable>(this)](
                const signals::connection &conn, auto... args) mutable {
                auto aw = std::move(awaited_from);
                conn.disconnect();
                result_cache.emplace(std::move(
                    args)...); // the result_catch lives in the coro frame
                std::move(aw).resume(); // release ownership & resume
            });
    }

    auto await_resume() // return the value
    {
        constexpr std::size_t size = std::tuple_size_v<args_type>;
        if constexpr (size == 1u) // single argument doesn't need a tuple
            return std::get<0u>(*std::move(result_cache));
        else if constexpr (size > 1u) // make a tuple if more than one arg
            return *std::move(result_cache);
        // else return void.
    }

    // capture it for lazy initialization
    Signal &signal;
    // capture the ownership of the awaiting coroutine
    cobalt::unique_handle<void> awaited_from;
    // store the result from the call
    std::optional<args_type> result_cache;

    // to manage shared ownership with an internal counter.
    // If the last gets released before the handler is invoked,
    // the coro just gets destroyed.
    std::size_t use_count{0u};
    friend void intrusive_ptr_add_ref(signal_awaitable *st) { st->use_count++; }
    friend void intrusive_ptr_release(signal_awaitable *st) {
        if (st->use_count-- == 1u)
            st->awaited_from.reset();
    }
};

namespace boost::signals2 {

// make all signals awaitable
template <typename... Args>
auto operator co_await(signals::signal<Args...> &sig)
    -> signal_awaitable<signals::signal<Args...>> {
    return {sig};
}

} // namespace boost::signals2

namespace asio = boost::asio;

cobalt::task<void> co_sleep(std::chrono::milliseconds t) {
    asio::steady_timer tim{co_await asio::this_coro::executor, t};
    co_await tim.async_wait(cobalt::use_op);
    co_return;
}

class curl_services {
  public:
    std::queue<signals::signal<void(bool)> *> push_sigs;
    cobalt::task<void> push(request &r) {
        if (request_queue.size() >= 1024) {
            signals::signal<void(bool)> sig;
            push_sigs.push(&sig);
            co_await sig;
        }
        request_queue.push(r);
        bool expected = false;
        if (loop_running.compare_exchange_strong(expected, true))
            curl_request_loop();
        co_return;
    }

    curl_services() { multi = curl_multi_init(); }
    ~curl_services() { curl_multi_cleanup(multi); }

  private:
    std::optional<request> pop() {
        if (request_queue.empty()) {
            return std::optional<request>{};
        }
        auto r = request_queue.front();
        request_queue.pop();
        if (push_sigs.size() > 0) {
            auto sig = push_sigs.front();
            (*sig)(true);
        }
        return r;
    }
    std::queue<request> request_queue;
    std::atomic_bool loop_running{false};
    CURLM *multi;
    cobalt::detached curl_request_loop() {
        std::cout << "start loop" << std::endl;
        int max_handles = 1024;
        int running_handles = 0;
        std::map<CURL *, request> running_requests;
        while (true) {
            if (running_handles == 0 && request_queue.empty()) {
                break;
            } else if (running_handles < max_handles) {
                for (int i = running_handles; i < max_handles; ++i) {
                    auto r = pop();
                    if (!r.has_value()) {
                        break;
                    }
                    curl_multi_add_handle(multi, r.value().handle);
                    running_requests.insert({r.value().handle, r.value()});
                }
            }
            CURLMcode mc = curl_multi_perform(multi, &running_handles);

            if (mc) {
                continue;
            }
            if (!mc && running_handles) {
                int n_ac_fd;
                mc = curl_multi_poll(multi, NULL, 0, 1, &n_ac_fd);
                if (n_ac_fd)
                    co_await co_sleep(std::chrono::milliseconds(10));
            }
            if (mc) {
                continue;
            }
            struct CURLMsg *m;
            do {
                int msgq = 0;
                m = curl_multi_info_read(multi, &msgq);
                if (!m || (m->msg != CURLMSG_DONE)) {
                    continue;
                }
                CURL *e = m->easy_handle;
                curl_multi_remove_handle(multi, e);
                if (running_requests.contains(e) && running_requests[e].chan)
                    co_await running_requests[e].chan->write(m->data.result);

            } while (m);
        }
        loop_running = false;
        std::cout << "stop loop" << std::endl;
    }
};

// not thread safe
template <typename T> class awaitable_queue {
  public:
    awaitable_queue(std::size_t size) : _size(size) {}
    cobalt::task<std::optional<T>> read() {
        if (!queue.empty()) {
            // there is data in queue read directly
            auto v = queue.front();
            queue.pop();
            if (!_closed && !write_sigs.empty()) {
                // wakeup the waiting write
                auto w_s = write_sigs.front();
                write_sigs.pop();
                (*w_s)(true);
            }
            co_return v;
        } else {
            if (_closed) {
                co_return {};
            }
            // no date in queue, waiting for write to wakeup us
            signals::signal<void(std::optional<T>)> r_sig;
            read_sigs.push(&r_sig);
            co_return co_await r_sig;
        }
    }

    cobalt::task<void> write(T v) {
        if (_closed) {
            co_return;
        }
        if (!read_sigs.empty()) {
            // there is waiting read, send it to read directly
            auto sig = read_sigs.front();
            read_sigs.pop();
            (*sig)(v);
            co_return;
        }
        if (queue.size() < _size) {
            // queue not full push directly
            queue.push(v);
        } else {
            // queue full, save write action to write write_sigs, wait for read
            // to wakeup up
            signals::signal<void(bool)> w_s;
            write_sigs.push(&w_s);
            auto closed = co_await w_s;
            if (!closed) {
                queue.push(v);
            }
            co_return;
        }
        co_return;
    }

    void close() {
        _closed = true;
        while (!read_sigs.empty()) {
            auto sig = read_sigs.front();
            read_sigs.pop();
            (*sig)(std::optional<T>{});
        }

        while (!write_sigs.empty()) {
            auto sig = write_sigs.front();
            write_sigs.pop();
            (*sig)(false);
        }
    }

    bool is_close() { return _closed; }

  private:
    std::queue<signals::signal<void(std::optional<T>)> *> read_sigs;
    std::queue<signals::signal<void(bool)> *> write_sigs;
    std::queue<T> queue;
    std::size_t _size;
    bool _closed{false};
};

size_t write_callback(char *ptr, size_t size, size_t nmemb, void *userdata) {
    return size * nmemb;
}
class rest_client {
  public:
    cobalt::task<int> get(const std::string &url) {
        cobalt::channel<CURLcode> chan;
        request r(curl_easy_init(), &chan);
        curl_easy_setopt(r.handle, CURLOPT_URL, url.c_str());
        curl_easy_setopt(r.handle, CURLOPT_URL, url.c_str());
        curl_easy_setopt(r.handle, CURLOPT_SSL_VERIFYHOST, 0);
        curl_easy_setopt(r.handle, CURLOPT_SSL_VERIFYPEER, 0);
        curl_easy_setopt(r.handle, CURLOPT_WRITEFUNCTION, write_callback);
        co_await cs.push(r);
        int code = co_await chan.read();
        co_return code;
    }

  private:
    thread_local static curl_services cs;
};

thread_local curl_services rest_client::cs;

cobalt::detached task_co(std::string url,
                         std::function<void(void)> complete_cb) {
    rest_client r;
    auto ret = co_await r.get(url);
    co_await co_sleep(std::chrono::seconds(3));
    std::cout << url << " ret:" << ret << std::endl;
    complete_cb();
}

std::vector<std::string> tasks = {
    "https://www.test.com",
    "https://www.baidu.com",
    "https://www.163.com",
    "https://www.aliyun.com",
};

auto pop_task() -> std::optional<std::string> {
    if (tasks.empty()) {
        return {};
    }
    auto url = tasks.back();
    tasks.pop_back();
    return {url};
};

cobalt::task<int> test() {
    int n = 0;
    int mn = 100;
    cobalt::channel<bool> idle_sig;
    cobalt::channel<bool> complete_sig;
    bool no_new_task = false;
    while (true) {
        auto ourl = pop_task();
        if (!ourl.has_value()) {
            idle_sig.close();
            no_new_task = true;
            std::cout << "no_new_task" << std::endl;
            break;
        }
        ++n;
        std::cout << ourl.value() << std::endl;
        task_co(ourl.value(), [&]() -> cobalt::detached {
            --n;
            if (idle_sig.is_open())
                co_await idle_sig.write(true);
            if (no_new_task) {
                std::cout << "complete_sig" << std::endl;
                co_await complete_sig.write(true);
            }
        });
        if (n < mn) {
            continue;
        }
        co_await idle_sig.read();
    }
    std::cout << "complete_sig r" << std::endl;
    co_await complete_sig.read();
    std::cout << "complete_sig r1" << std::endl;
    co_return 0;
}

int main(int argc, char *argv[]) { return cobalt::run(test()); }